package com.chiancloud.shield.core.main;

import com.google.common.base.Preconditions;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.util.ExitUtil;
import org.apache.hadoop.util.ShutdownHookManager;
import org.apache.hadoop.util.VersionInfo;
import org.apache.hadoop.yarn.YarnUncaughtExceptionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.net.MalformedURLException;
import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by wanghui on 14-8-20.
 */
public class ServiceLauncher <S extends Service>
        implements LauncherExitCodes, IrqHandler.Interrupted,
            Thread.UncaughtExceptionHandler {

    private static final Logger LOG = LoggerFactory.getLogger(
            ServiceLauncher.class);

    protected static final int PRIORITY = 30;

    public static final String NAME = "ServiceLauncher";



    /**
     * Name of the "--conf" argument.
     */
    public static final String ARG_CONF = "--conf";

    public static final String USAGE_MESSAGE =
            "Usage: " + NAME + " classname ["+ARG_CONF + "<conf file>] <service arguments> | ";
    static final int SHUTDOWN_TIME_ON_INTERRUPT = 30 * 1000;

    private volatile S service;
    private int serviceExitCode;
    private final List<IrqHandler> interruptHandlers = new ArrayList<IrqHandler>(1);
    private Configuration configuration;
    private String serviceClassName;
    private static AtomicBoolean signalAlreadyReceived = new AtomicBoolean(false);

    public ServiceLauncher(String serviceClassName) {
        this.serviceClassName = serviceClassName;
    }


    public static void serviceMain(List<String> argsList) {
        if (argsList.isEmpty()) {
            exitWithMessage(EXIT_USAGE, USAGE_MESSAGE);
        } else {
            String serviceClassName = argsList.get(0);

            if (LOG.isDebugEnabled()) {
                LOG.debug(startupShutdownMessage(serviceClassName, argsList));
                StringBuilder builder = new StringBuilder();
                for (String arg : argsList) {
                    builder.append('"').append(arg).append("\" ");
                }
                LOG.debug(builder.toString());
            }
            Thread.setDefaultUncaughtExceptionHandler(
                    new YarnUncaughtExceptionHandler());

            ServiceLauncher serviceLauncher = new ServiceLauncher<Service>(serviceClassName);
            serviceLauncher.launchServiceAndExit(argsList);
        }
    }


    private static void exitWithMessage(int status, String message) {
        System.err.println(message);
        ExitUtil.terminate(status);
    }
    public static String startupShutdownMessage(String classname,
                                                List<String> args) {
        final String hostname = NetUtils.getHostname();

        return toStartupShutdownString("STARTUP_MSG: ", new String[]{
                "Starting " + classname,
                "  host = " + hostname,
                "  args = " + args,
                "  version = " + VersionInfo.getVersion(),
                "  classpath = " + System.getProperty("java.class.path"),
                "  build = " + VersionInfo.getUrl() + " -r "
                        + VersionInfo.getRevision()
                        + "; compiled by '" + VersionInfo.getUser()
                        + "' on " + VersionInfo.getDate(),
                "  java = " + System.getProperty("java.version")
        });
    }
    private static String toStartupShutdownString(String prefix, String[] msg) {
        StringBuilder b = new StringBuilder(prefix);
        b.append("\n/************************************************************");
        for (String s : msg) {
            b.append("\n").append(prefix).append(s);
        }
        b.append("\n************************************************************/");
        return b.toString();
    }
    public void launchServiceAndExit(List<String> args) {
        registerInterruptHandler();
        Configuration conf = new Configuration();
        String[] processedArgs = extractConfigurationArgs(conf, args);
        ExitUtil.ExitException ee = launchServiceRobustly(conf, processedArgs);
        exit(ee);
    }
    protected void registerInterruptHandler() {
        try {
            interruptHandlers.add(new IrqHandler(IrqHandler.CONTROL_C, this));
            interruptHandlers.add(new IrqHandler(IrqHandler.SIGTERM, this));
        } catch (IOException e) {
            error("Signal handler setup failed : {}" + e, e);
        }
    }
    public static String[] extractConfigurationArgs(Configuration conf,
                                                    List<String> args) {

        //convert args to a list
        int argCount = args.size();
        if (argCount <= 1 ) {
            return new String[0];
        }
        List<String> argsList = new ArrayList<String>(argCount);
        ListIterator<String> arguments = args.listIterator();
        arguments.next();
        while (arguments.hasNext()) {
            String arg = arguments.next();
            if (arg.equals(ARG_CONF)) {

                if (!arguments.hasNext()) {
                    exitWithMessage(EXIT_COMMAND_ARGUMENT_ERROR,
                            ARG_CONF + ": missing configuration file after ");
                }
                File file = new File(arguments.next());
                if (!file.exists()) {
                    exitWithMessage(EXIT_COMMAND_ARGUMENT_ERROR,
                            ARG_CONF + ": configuration file not found: " + file);
                }
                try {
                    conf.addResource(file.toURI().toURL());
                } catch (MalformedURLException e) {
                    exitWithMessage(EXIT_COMMAND_ARGUMENT_ERROR,
                            ARG_CONF + ": configuration file path invalid: " + file);
                }
            } else {
                argsList.add(arg);
            }
        }
        String[] processedArgs = new String[argsList.size()];
        argsList.toArray(processedArgs);
        return processedArgs;
    }

    public ExitUtil.ExitException launchServiceRobustly(Configuration conf,
                                                        String[] processedArgs) {
        ExitUtil.ExitException exitException;
        try {
            int exitCode = launchService(conf, processedArgs, true);
            if (service != null) {
                Throwable failure = service.getFailureCause();
                if (failure != null) {
                    Service.STATE failureState = service.getFailureState();
                    if (failureState == Service.STATE.STOPPED) {
                        LOG.debug("Failure during shutdown:{} ", failure, failure);
                    } else {
                        throw failure;
                    }
                }
            }
            exitException = new ExitUtil.ExitException(exitCode,
                    "In " + serviceClassName);
        } catch (ExitUtil.ExitException ee) {
            exitException = ee;
        } catch (Throwable thrown) {
            exitException = convertToExitException(thrown);
        }
        return exitException;
    }

    protected ExitUtil.ExitException convertToExitException(Throwable thrown) {
        ExitUtil.ExitException exitException;
        int exitCode;
        String message = thrown.getMessage();
        if (message == null) {
            message = thrown.toString();
        }
        if (thrown instanceof ExitCodeProvider) {
            exitCode = ((ExitCodeProvider) thrown).getExitCode();
            if (LOG.isDebugEnabled()) {
                LOG.debug("While running {}: {}", getServiceName(), message, thrown);
            }
            LOG.error(message);
        } else {
            // not any of the service launcher exceptions -assume something worse
            error(message, thrown);
            exitCode = EXIT_EXCEPTION_THROWN;
        }
        exitException = new ExitUtil.ExitException(exitCode, message);
        exitException.initCause(thrown);
        return exitException;
    }

    public int launchService(Configuration conf,
                             String[] processedArgs,
                             boolean addProcessHooks)
            throws Throwable {

        instantiateService(conf);

        // add any process shutdown hooks
        if (addProcessHooks) {
            ServiceShutdownHook shutdownHook = new ServiceShutdownHook(service);
            ShutdownHookManager.get().addShutdownHook(shutdownHook, PRIORITY);
        }
        RunService runService = null;

        if (service instanceof RunService) {
            runService = (RunService) service;
            configuration = runService.bindArgs(configuration, processedArgs);
            Preconditions.checkNotNull(configuration,
                    "null configuration returned by bindArgs()");
        }

        if (!service.isInState(Service.STATE.INITED)) {
            service.init(configuration);
        }
        service.start();
        int exitCode = EXIT_SUCCESS;
        if (runService != null) {
            //assume that runnable services are meant to run from here
            exitCode = runService.runService();
            LOG.debug("Service exited with exit code {}", exitCode);

        } else {
            LOG.debug("waiting for service threads to terminate");
            service.waitForServiceToStop(0);
        }
        //exit
        serviceExitCode = exitCode;
        return serviceExitCode;
    }

    public Service instantiateService(Configuration conf)
            throws ClassNotFoundException, InstantiationException, IllegalAccessException,
            ExitUtil.ExitException, NoSuchMethodException, InvocationTargetException {
        Preconditions.checkArgument(conf != null, "null conf");
        configuration = conf;

        Class<?> serviceClass =
                this.getClass().getClassLoader().loadClass(serviceClassName);
        Object instance = serviceClass.getConstructor().newInstance();
        if (!(instance instanceof Service)) {
            throw new ExitUtil.ExitException(EXIT_BAD_CONFIGURATION,
                    "Not a Service class: " + serviceClassName);
        }

        service = (S) instance;
        return service;
    }

    public String getServiceName() {
        Service s = service;
        String name = null;
        if (s != null) {
            try {
                name = s.getName();
            } catch (Exception ignored) {
                // ignored
            }
        }
        if (name != null) {
            return "service " + name;
        } else {
            return "service classname " + serviceClassName;
        }
    }


    protected void warn(String text) {
        System.err.println(text);
    }

    protected void error(String message, Throwable thrown) {
        String text = "Exception: " + message;
        warn(text);
        LOG.error(text, thrown);
    }

    protected void exit(int exitCode, String message) {
        ExitUtil.terminate(exitCode, message);
    }

    protected void exit(ExitUtil.ExitException ee) {
        ExitUtil.terminate(ee.status, ee);
    }

    @Override
    public void interrupted(IrqHandler.InterruptData interruptData) {
        String message = "Service interrupted by " + interruptData.toString();
        warn(message);
        if (!signalAlreadyReceived.compareAndSet(false, true)) {
            warn("Repeated interrupt: escalating to a JVM halt");
            ExitUtil.halt(EXIT_INTERRUPTED, message);
        }
        int shutdownTimeMillis = SHUTDOWN_TIME_ON_INTERRUPT;
        ServiceForcedShutdown forcedShutdown =
                new ServiceForcedShutdown(shutdownTimeMillis);
        Thread thread = new Thread(forcedShutdown);
        thread.setDaemon(true);
        thread.start();
        try {
            thread.join(shutdownTimeMillis);
        } catch (InterruptedException ignored) {
        }
        if (!forcedShutdown.isServiceStopped()) {
            warn("Service did not shut down in time");
        }
        exit(EXIT_INTERRUPTED, message);
    }

    @Override
    public void uncaughtException(Thread thread, Throwable exception) {
        if (ShutdownHookManager.get().isShutdownInProgress()) {
            LOG.error("Thread {} threw an error during shutdown: {}.",
                    thread.toString(),
                    exception,
                    exception);
        } else if (exception instanceof Error) {
            try {
                LOG.error("Thread {} threw an error: {}. Shutting down",
                        thread.toString(),
                        exception,
                        exception);
            } catch (Throwable err) {

            }
            if (exception instanceof OutOfMemoryError) {
                try {
                    System.err.println("Halting due to Out Of Memory Error...");
                } catch (Throwable err) {
                }
                ExitUtil.halt(EXIT_EXCEPTION_THROWN);
            } else {
                exit(convertToExitException(exception));
            }
        } else {
                LOG.error("Thread {} threw an exception: {}",
                    thread.toString(),
                    exception,
                    exception);
        }
    }
    protected class ServiceForcedShutdown implements Runnable {

        private final int shutdownTimeMillis;
        private boolean serviceStopped;

        public ServiceForcedShutdown(int shutdownTimeoutMillis) {
            this.shutdownTimeMillis = shutdownTimeoutMillis;
        }

        @Override
        public void run() {
            if (service != null) {
                service.stop();
                serviceStopped = service.waitForServiceToStop(shutdownTimeMillis);
            } else {
                serviceStopped = true;
            }
        }

        private boolean isServiceStopped() {
            return serviceStopped;
        }
    }
}
